package com.pf.messageQueue.common;

import com.pf.messageQueue.service.RedisMessageQueueService;
import com.pf.utils.SpringContextUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

/**
 * redis list 消息消费守护线程
 */
@Slf4j
public class MessageQueueConsumerMonitor {

    private Thread consumerThread;
    private volatile boolean toStop = false;
    public static MessageQueueConsumerMonitor monitor = new MessageQueueConsumerMonitor();

    private RedisMessageQueueService redisMessageQueueService = SpringContextUtils.getBean(RedisMessageQueueService.class);


    public static MessageQueueConsumerMonitor getInstance() {
        return monitor;
    }

    public void start() {
        consumerThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!toStop) {
                    try {

                        // 消费消息
                        redisMessageQueueService.rpop();

                    } catch (Exception e) {
                        if (!toStop) {
                            log.error(e.getMessage(), e);
                        }
                    }

                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        if (!toStop) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });
        consumerThread.setDaemon(true);
        consumerThread.setName("consumerThread");
        consumerThread.start();
    }

    /**
     * 关闭
     */
    public void toStop() {
        toStop = true;
        if (consumerThread != null) {
            consumerThread.interrupt();
            try {
                consumerThread.join();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }
        }
    }
}
